import java.util.concurrent.Semaphore;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * A solution to the producer-consumer problem using a shared buffer. This one
 * uses counting semaphores to keep track of the number of empty and the number
 * of full slots.
 * 
 * @author Barbara Lerner
 * @version September 22. 2010
 * 
 */
public class ProducerConsumerUsingSemaphore {
	// Size of the buffer
	private static final int BUFFER_SIZE = 10000;

	// The buffer that values are written to and read from
	private Integer[] buffer = new Integer[BUFFER_SIZE];

	// Number of elements in the buffer.
	private int count = 0;

	// Slot where the next value should be written
	private int in = 0;

	// Slot where the next value should be read from
	private int out = 0;

	// Lock to control entry to the critical section.
	private Lock lock = new ReentrantLock();

	// Counting semaphore for the number of full slots.
	private Semaphore full = new Semaphore(0);

	// Counting semaphore for the number of empty slots.
	private Semaphore empty = new Semaphore(BUFFER_SIZE);

	/**
	 * Add a value to the buffer. If the buffer is full, it waits until there is
	 * room in the buffer
	 * 
	 * @param item
	 *            the value to add
	 * @throws InterruptedException
	 *             if the thread is interrupted while it is waiting for a slot
	 *             to fill.
	 */
	public void produce(Integer item) throws InterruptedException {
		// Use up one of the empty slots.
		empty.acquire();

		// Get the mutex lock.
		lock.lock();

		try {
			System.out.println("Producer in critical section.");
			System.out.flush();

			// Make sure the slot is empty. This is a sanity check!
			if (buffer[in] != null) {
				System.out.println("ACK!  There is already something here!");
				System.out.flush();
			}

			// Add an item to the buffer
			buffer[in] = item;
			in = (in + 1) % BUFFER_SIZE;
			count++;

			System.out.println("Producer done with critical section.");
			System.out.flush();
		} finally {
			lock.unlock(); // Ok for others to enter now.
		}

		// There is another filled slot.
		full.release();
	}

	/**
	 * Remove an item from the buffer. Waits if the buffer is empty.
	 * 
	 * @return the item removed
	 * @throws InterruptedException
	 */
	public Integer consume() throws InterruptedException {
		// Grab one of the full slots.
		full.acquire();

		// Grab the mutex lock.
		lock.lock();

		Integer item;
		try {
			System.out.println("Consumer in critical section.");
			System.out.flush();

			item = buffer[out];

			// Sanity check that there is actually something there!
			if (item == null) {
				System.out.println("ACK!  The slot is empty!");
				System.out.flush();
			}

			// remove an item from the buffer
			buffer[out] = null;
			out = (out + 1) % BUFFER_SIZE;
			count--;

			System.out.println("Consumer done with critical section.");
			System.out.flush();
		} finally {
			lock.unlock(); // Ok for others to enter now.
		}
		empty.release(); // There is another empty slot available.

		return item;
	}

	/**
	 * Run a producer and consumer, producing and consuming the same number of
	 * items. The buffer should be empty at the end. Print a message indicating
	 * if that is so or not.
	 * 
	 * @param args
	 *            none used
	 */
	public static void main(String[] args) {
		// Number of items to produce and consume
		final int numReps = 10000;

		// The shared buffer
		final ProducerConsumerUsingSemaphore buffer = new ProducerConsumerUsingSemaphore();

		// The thread that will write to the buffer
		Thread producer = new Thread(new Runnable() {
			public void run() {
				try {
					for (int i = 1; i < numReps; i++) {
						buffer.produce(new Integer(i));
					}
				} catch (InterruptedException e) {
					System.out.println("Producer was interrupted.");
				}
			}
		});

		// The thread that will read from the buffer
		Thread consumer = new Thread(new Runnable() {
			public void run() {
				try {
					for (int i = 1; i < numReps; i++) {
						buffer.consume();
					}
				} catch (InterruptedException e) {
					System.out.println("Consume was interrupted.");
				}
			}
		});

		// Start the producer and consumer. They will run concurrently
		producer.start();
		consumer.start();

		try {
			// Wait for both the producer and consumer to complete.
			producer.join();
			consumer.join();

			// Check if the buffer is empty.
			if (buffer.count == 0) {
				System.out.println("Buffer empty.  Good!");
			} else {
				System.out.println("ACK!  There are " + buffer.count
						+ " values in the array.");
			}

		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}